package com.qyer.search.storm.bolt;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import backtype.storm.utils.TupleUtils;
import com.qyer.search.storm.util.ConfigUtils;
import com.qyer.search.storm.util.EntityTags;
import com.qyer.search.storm.util.Tag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.util.parsing.combinator.testing.Str;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by tangzheng on 16/1/8.
 */
public class FetchTagBolt extends JDBCBolt {
  private static final long serialVersionUID = -1L;
  //private static final Logger LOG = LoggerFactory.getLogger(FetchTagBolt.class);

  private String _streamId;
  public FetchTagBolt(String tagBoltPrefix, int inCount, int inCount2, String streamId,
                      Map<String, String> sqlConfig){
    super(tagBoltPrefix, inCount, inCount2, sqlConfig);
    this._streamId = streamId;
  }

  @Override public void execute(Tuple tuple) {
    //currentTime(_prefix + "execute in,");
    if (TupleUtils.isTick(tuple)) {
      //LOG.info("Tick arrive.");
      tickInfoLogger.info("Tick arrive.");
      checkDBConnection();
    }else {
      checkDBConnection();

      long now = System.currentTimeMillis();

      String index = tuple.getString(0);
      String type = tuple.getString(1);
      List<Integer> payload = (List<Integer>) tuple.getValue(2);
      int msgCount = tuple.getInteger(3);
      int stage = tuple.getInteger(4);
      PreparedStatement pstmt_in_use;
      int inCount_in_use = 0;
      //******************************************************DB
      /*
      LOG.info("--------------------fetchTagIn:{}----------------------->", msgCount);
      LOG.info("payload:{}", payload);
      LOG.info("index  :{}", index);
      LOG.info("type   :{}", type);
      LOG.info("msgId  :{}", msgCount);
      LOG.info("stage  :{}", stage);
      LOG.info("<------------------fetchTagIn:{}-------------------------", msgCount);
      */
      //********************************************************************
      dataLog.info("{},msgId:{},index:{},type:{},stage:{},payload:{}", ConfigUtils.IN, msgCount, index, type, stage, payload);
      //********************************************************************

      pstmt_in_use = (ConfigUtils.INIT_STAGE == stage) ? _pstmt : _pstmt2;
      inCount_in_use = (ConfigUtils.INIT_STAGE == stage) ? _inCount : _inCount2;

      try {
        ConfigUtils.constructSQL(pstmt_in_use, payload, inCount_in_use);
      } catch (Exception e) {
        //LOG.error("constructSQL failed,{}", e);
        dbInfoLogger.info("msgId:{},constructSQL failed,{}", msgCount, e);
      }

      int lid;
      int tid;
      String tag_name;
      int tag_type;
      double coefficient;
      int status;
      Map<Integer, EntityTags> tagsMap = new HashMap<>();

      try {
        ResultSet rs = pstmt_in_use.executeQuery();
        while (rs.next()) {
          lid = rs.getInt(1);
          tid = rs.getInt(2);
          tag_name = rs.getString(3);
          tag_type = rs.getInt(4);
          coefficient = rs.getDouble(5);
          status = rs.getInt(6);

          Tag tag = new Tag(tid, tag_name, tag_type, coefficient, status);

          //LOG.info("msgCount:{},product id:{},tag:{}", msgCount, lid, tag.toString());
          //infoLogger.info("msgCount:{},product id:{},tag:{}", msgCount, lid, tag.toString());
          processLogger.info("msgId:{},product id:{},tid:{},tag_name:{},tag_type:{},coefficient:{},status:{}",
                             msgCount, lid, tid, tag_name, tag_type, coefficient, status);

          if (tagsMap.containsKey(lid)) {
            tagsMap.get(lid).addTags(tag);
          } else {
            List<Tag> lt = new ArrayList<>();
            lt.add(tag);
            EntityTags et = new EntityTags(lid, lt);
            tagsMap.put(lid, et);
          }
        }

        //currentTime(_prefix + "execute db out,");
        if(tagsMap == null){
          tagsMap = new HashMap<>();
        }
        ////////////////////////////////////////
        long dbTime = System.currentTimeMillis();
        int dbCost = (int)(dbTime - now);
        ///////////////////////////////////////

        String tagsStr = ConfigUtils.DEFAULT_GSON.toJson(tagsMap);
        //currentTime(_prefix + ",execute out,");
        _outputCollector.emit(_streamId, tuple, new Values(index, type, tagsStr, msgCount));
        /*
        LOG.info("msgCount:{},payload:{},tagmap:{}", msgCount, payload, tagsStr);
        LOG.info("---------------------fetchTagOut:{}-----------------------", msgCount);
        LOG.info("index  :{}", index);
        LOG.info("type   :{}", type);
        LOG.info("tagsStr  :{}", tagsStr);
        LOG.info("msgCount  :{}", stage);
        LOG.info("---------------------fetchTagOut:{}-----------------------", msgCount);*/

        ////////////////////////////////////////
        dataLog.info("{},msgId:{},index:{},type:{},stage:{},payload:{}", ConfigUtils.OUT, msgCount, index, type, stage, tagsStr);
        ////////////////////////////////////////
        _outputCollector.ack(tuple);

        ////////////////////////////////////////
        long end = System.currentTimeMillis();
        int cost = (int)(end - now);
        ////////////////////////////////////////

        timeCostLogger.info("{}:{}:{}:{}:{}:{}", msgCount, now, dbTime, end, dbCost, cost);

      } catch (Exception e) {
        //LOG.error("DB get tags failed,{}", e);
        //LOG.error("msgCount:{},payload:{}", msgCount, payload);
        errorLogger.info("msgId:{},DB get tags failed,{}", msgCount, e);
        errorLogger.info("msgId:{},payload:{}", msgCount, payload);

        _outputCollector.fail(tuple);
      }
      //******************************************************DB

      //currentTime(_prefix + "execute out,");
    }
  }

  @Override public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    outputFieldsDeclarer.declareStream(_streamId, new Fields("index", "type", "tags", "msgId"));
  }


}
